<?php

/*This script requires PEAR Net_Gearman */
require_once 'Net/Gearman/Client.php';
require_once 'php-sql-parser.php';
require_once 'shard-key-mapper.php';

class OHIPShardQuery{
	var $messages = array();
	var $push_where="";
	var $inlist_merge_threshold = 500;
	var $inlist_merge_size = 128;
	var $engine='MEMORY';

	var $cols;
	var $debug;
	var $fetch_method;
	var $gearman_servers;
	var $inlist_opt;
	var $row_count;
	var $table_name;
	var $tmp_shard;
	var $verbose;

	var $mysqli = null; #mysql conn

	var $workers; # the count of workers

	var $force_shard = array();
	var $error = false;
	var $partition_callback = false;
	var $partition_column = false;
	var $coord_odku = array();
	var $subqueries = array();
	var $uuid;
	
	
	//Construct
	function __construct($shards, $params = array()) {
		$this->uuid = uniqid();
		ini_set('memory_limit', '1024M');
		$this->params = $params;
		$this->verbose = false;
		$this->fetch_method = "query";
		$this->inlist_opt = false;
		$this->between_opt = false;
		$this->gearman_servers = array();
		$this->debug = false;
		$this->row_count = 0;
		$this->tmp_shard = array();
		$this->max_degree = false;
		$shared = false;

		if(in_array('verbose', array_keys($params))) {
			$this->verbose = true;
		}

		$this->max_degree = count($shards) * 2;

		if(!empty($params['max_degree'])) $this->max_degree = $params['max_degree'];
	
		#merge any explicit parameters with the defaults and/or INI options
		#$params = array_merge(get_shards($params, true), $params);

		if(!empty($params['mapper']) && $params['mapper'] == 'directory') {
			if(empty($params['directory'])) $params['directory'] = 'directory';
			if(!empty($params['ignore_shard'])) $params['ignore_shard'] .= ',';
			$params['ignore_shard'] .= 'directory';
		}
			
		if(!empty($params['shared'])){
			$shared = $shards[$params['shared']];
			if(!empty($params['ignore_shard'])) $params['ignore_shard'] .= ',';
			$params['ignore_shard'] .= $params['shared'];
		}

		if(!empty($params['push_where'])) {
			$this->push_where = $params['push_where'];
		}

		if(!empty($params['inlist_merge_threshold'])) {
			$this->inlist_merge_threshold = $params['inlist_merge_threshold'];
		}

		if(!empty($params['inlist_merge_size'])) {
			$this->inlist_merge_size = $params['inlist_merge_size'];
		}

		if(!empty($params['coord_engine'])) {
			$this->engine = $params['coord_engine'];
		}

		#unset any shards we want to ignore (normally just directory)
		if(!empty($params['ignore_shard'])) {
			$ignore = explode(',',$params['ignore_shard']);
			foreach($ignore as $shard) {
				unset($shards[$shard]);
			}
		}

		$mapper=false; 
		if(!empty($params['mapper'])) {
			switch ($params['mapper']) {
				case 'hash':
					$mapper = new HashShardKeyMapper($shards);
					$this->shared = $shared;
				break;

				case 'directory':
					$mapper = new DirectoryShardKeyMapper($directory);
					$shards = array_merge($shards,$map->get_shards());
					$this->shared = $shared;
				break;

				default:
				break;
			}

			if(!empty($params['column'])) $this->set_partition_info($params['column'], array($mapper,'map'));
		}


		if(!empty($params['chunk'])) $this->chunk = $params['chunk']; else $this->chunk = false;
	
		$this->shard_ids = array_keys($shards);
		$this->shards = $shards;
		
		/* One shard acts as the coordination shard.  A table (or a temporary table) is created on this shard
		   to coallesce the results from the other shards.  This shard is used when a query broadcast
		   is necessary, otherwise only one shard will be used.
		*/
		if(!empty($params['coord_shard'])) {
			if(is_array($params['coord_shard'])) {
				$this->tmp_shard = $params['coord_shard'];
			} else {
				$this->tmp_shard = $shards[$params['coord_shard']];
			}
		} else {
			#pick a random shard
			$this->tmp_shard = $shards[array_rand($shards)];
		}
	
	/* Only one method
		if(!empty($params['method'])) {
			if($params['method'] == 'fetch') {
				$this->fetch_method = 'fetch_resultset';
			}
			if($params['method'] == 'store') {
				$this->fetch_method = 'store_resultset';
			}
		}
	*/
	
		if(!empty($params['inlist'])) {
			$this->inlist_opt = trim($params['inlist']);
		}

		if(!empty($params['between'])) {
			$this->between_opt = trim($params['between']);
		}

		if(!empty($params['gearman'])) {
			$this->gearman_servers = explode(',', $params['gearman']);
		} else {
			$this->gearman_servers = array('localhost:4730');
		}

		//mysql conn
		$this->mysqli = new mysqli($this->tmp_shard['host'], $this->tmp_shard['user'], $this->tmp_shard['password'], $this->tmp_shard['db']);	
		if ($this->mysqli->connect_error) {
			throw new Exception("connect to the coordinator DB:" . $this->mysqli->connect_error . "\n");
		}
				
	}
	

	#this function works only with SELECT statements and it only works with a single query
	#the query may be parallelized based on the --inlist, --between options and the number of shards
	function query($sql) {
	
		$this->insert_sql = "";
		$this->insert_values = array();
	
		$start = microtime(true);
		
		//init connection to coord shard
		//var_dump($this->tmp_shard); exit;

		
		//connect to tmp coord shard?
		/*
		$this->conn = $this->my_connect($this->tmp_shard);
		
		if(!$this->conn) {
			throw new Exception("connect to the coordinator DB:" . $this->my_error() . "\n");
		}
		
		if(!$this->my_select_db($this->tmp_shard['db'], $this->conn)) {
			throw new Exception("use the coord db database:" . $this->my_error($this->conn));
		} */
		

		//connect to gearmand server
		$this->workers = 0;
		$this->client = new Net_Gearman_Client($this->gearman_servers) or die('Could not instantiate gearman client\n');

		$this->subqueries = array();

		
		//process sql
		if(!$this->process_sql($sql)) {
			return false;	
		}
		
		//alex
		//build shard arry
	//	foreach($this->shard_sql as $q){
		//	$this->shard_sql_new[] = array('sql' => $q , 'coord_odku' => $this->coord_odku, 'agg_key_cols' =>  $this->agg_key_cols);
	//	}
	//	$this->shard_sql = $this->shard_sql_new;

		$this->total_time = $this->parse_time = microtime(true) - $start;
		$start = microtime(true);


		//debug
		if($this->verbose){
		echo "-- SUBQUERIES\n";
		var_dump($this->subqueries);
		echo "-- SENDING TO SHARD\n";
		var_dump($this->shard_sql);
		echo "-- Agg table\n";
		var_dump($this->table_name);
		echo "-- coord\n";
		var_dump($this->coord_sql);
		echo "-- AGG KEYS COLS\n";
		var_dump($this->agg_key_cols);
		echo "-- ODKU\n";
		var_dump($this->coord_odku);
		}
		
		$collect_cb = array($this,'collect_rowset');
		$failure_cb = array($this,'handle_failure');

		#subqueries in the from clause get parallelized on each storage node
		if(count($this->subqueries) > 0 ) {
			if($this->verbose) {
				$this->messages[] = "Shard-Query has detected" . count($this->subqueries) . " subqueries in the FROM clause\n";
			}
		}

		//Send Sub Queries to Shards
		if(!empty($this->subqueries)) {
			$set = new Net_Gearman_Set();
		//	$save_agg_keys = $this->agg_key_cols;
		//	$save_odku = $this->coord_odku;
		//	$this->agg_key_cols = "";
		//	$this->coord_odku=array();
			foreach($this->subqueries as $table => $subsql) {
				#last param is a reference, so this will populate the set
				$this->create_gearman_set($subsql, $table, $collect_cb, $failure_cb,$set);
			}
		//	$this->agg_key_cols = $save_agg_keys;
		//	$this->coord_odku=$save_odku;
				
			
			$this->run_set($set);
			$msg = 0;
			if($this->workers>0 ) {
				sleep(.0005);
			}

			if($this->errors) {
				if($this->verbose)
					print_r($this->errors);
				return false;
			} 

		}

		//Send main queries to shards
		$set = new Net_Gearman_Set();
		$this->workers = 0;
		
		//$this->conn = $this->my_connect($this->tmp_shard);
		$this->create_gearman_set($this->shard_sql, $this->table_name, $collect_cb, $failure_cb, $set);
		$this->run_set($set);
		$spins =0 ;

		while($this->workers > 0) {
			sleep(.0005);
		}

		if($this->errors) {
			if($this->verbose)
				print_r($this->errors);
			return false;
		} 

		
		//Aggregate Results
		$stmt = $this->aggregate_result();
		
		//timing
		$this->exec_time = microtime(true) - $start;
		$this->total_time += $this->exec_time;

		//mysql_close($this->conn);

		return $stmt;
	}

	#this function broadcast one or more queries of any type to all shards
	function broadcast($queries,$collect_cb=null,$failure_cb=null) {
		if(!$collect_cb) $collect_cb=array($this,'generic_rowset');
		if(!$failure_cb) $failure_cb=array($this,'handle_failure');
		$start = microtime(true);
		$this->coord_sql = false;
		$this->force_shard = false;
		$this->table_name = false;
		$save_fetch_method = $this->fetch_method;
		
		$this->errors = false;	
		$result = $this->run_set($this->create_gearman_set($queries, false, $collect_cb, $failure_cb));
		$this->fetch_method = $save_fetch_method;
		$this->exec_time = microtime(true) - $start;
		return $result;
		
	}



	#FIXME: Make this support SUM(X) + SUM(Y).
	#To do this create process_select_expr and then iterate (possively recursively) 
	#over all expressions in the select calling that function
	protected function process_select($select, $straight_join=false) {
		$error = array();	
		$shard_query=""; 		#Query to send to each shard
		$coord_query=""; 		#Query to send to the coordination node
	
		$avg_count = 0;
		
		$group = array();		#list of positions which contain non-aggregate functions
		$push_group=array();  	#this is necessary for non-distributable aggregate functions
		$group_aliases=array(); #list of group aliases which will be indexed on the aggregation temp table
						  
		$is_aggregate = false;
		$coord_odku = array();

		
		$used_agg_func=0;

		$push_select = array(); #list of expressions to push onto the end of the SELECT clause sent to the shards
		$this->non_distrib = array(); #aliases of the non-distributable aggregate functions that will need to be "unpacked" before aggregation
		$this->orig_group[] = array(); #aliases of original (non-pushed) group-by expressions.  This is used to join the expressions later.

		foreach($select as $pos => $clause) {
			if($shard_query) $shard_query .= ",";	
			if($coord_query) $coord_query .= ",";	
		

			if(!empty($clause['base_expr']) && $clause['base_expr'] == "*") {
				$error[] = array('error_clause' => '*', 'error_reason'=>'"SELECT *" is not supported');
				continue;

			}
			$alias = $clause['alias'];

			if(strpos($alias,'.')) {
				$alias=trim($alias,'``');
				$alias = explode('.', $alias);
				$alias = $alias[1];
				$alias="`$alias`";
			}

			$base_expr = $clause['base_expr']; 

			switch($clause['expr_type']) {
				case 'expression':
					if($clause['sub_tree'][0]['expr_type'] == 'aggregate_function') {
						$is_aggregate = true;
						$skip_next = true;
						$base_expr = $clause['sub_tree'][1]['base_expr'];
						$alias = $clause['alias'];
						$function = $clause['sub_tree'][0]['base_expr'];
					
						switch($function) {
							case 'MIN':
							case 'MAX':
							case 'SUM':
								$used_agg_func=1;
								$expr_info=explode(" ", $base_expr);
								if(!empty($expr_info[0]) && strtolower($expr_info[0]) == 'distinct') {
										
										if($this->verbose) {
											echo "Detected a {$function} [DISTINCT] expression!\n";
										}
										
										unset($expr_info[0]);
										$new_expr = join(" ", $expr_info);
										$shard_query .= "$new_expr AS $alias";
										$coord_query .= "{$function}(distinct $alias) as $alias";	
										$push_group[] = $pos + 1;								
								} else {
									switch($function) {
										case 'SUM':
											$coord_odku[] = "$alias=$alias +  VALUES($alias)";
											break;
										case 'MIN':
											$coord_odku[] = "$alias=IF($alias < VALUES($alias), VALUES($alias),$alias)";
											break;
										case 'MAX':
											$coord_odku[] = "$alias=IF($alias > VALUES($alias), VALUES($alias), $alias)";
											break;
									}
									$shard_query .= "{$function}({$base_expr}) AS $alias";		
									$coord_query .= "{$function}({$alias}) AS $alias";
								}
								
							break;
							
							case 'AVG':								
								$used_agg_func=1;
								$expr_info=explode(" ", $base_expr);
								if(!empty($expr_info[0]) && strtolower($expr_info[0]) == 'distinct') {
										if($this->verbose) {
											echo "Detected a {$function} [DISTINCT] expression!\n";
										}
										
										unset($expr_info[0]);
										$new_expr = join(" ", $expr_info);
										$shard_query .= "$new_expr AS $alias";
										$coord_query .= "{$function}(distinct $alias) as $alias";
										#since we pushed a GB expression we need to update the ODKU clause
										$coord_odku[] = "$alias=VALUES($alias)";
										$push_group[] = $pos + 1;
										$group_aliases[] = $alias;
								} else {
										$alias_cnt = trim($alias,'`');
										$alias_cnt = "`{$alias_cnt}_cnt`";
										$shard_query .= "SUM({$base_expr}) AS {$alias}";		
										$coord_query .= "SUM({$alias})/SUM({$alias_cnt}) as {$alias}";
										
										$coord_odku[] = "$alias=$alias +  VALUES($alias)";

										#need to push a COUNT expression into the SELECT clause
										$push_select[] = "COUNT({$base_expr}) as {$alias_cnt}";
										$coord_odku[] = "{$alias_cnt}={$alias_cnt} +  VALUES({$alias_cnt})";

								}
								

							break;

							case '90TILE':
							case 'STDDEV':
							case 'STD':
							case 'STDDEV_POP':
							case 'STDDEV_SAMP':
							case 'VARIANCE':
							case 'VAR_POP':
							case 'VAR_SAMP':
							case 'GROUP_CONCAT':
								$used_agg_func=1;
								$expr_info=explode(" ", $base_expr);
								if(!empty($expr_info[0]) && strtolower($expr_info[0]) == 'distinct') {
										if($this->verbose) {
											echo "Detected a {$function} [DISTINCT] expression!\n";
										}
										
										unset($expr_info[0]);
										$new_expr = join(" ", $expr_info);
										$shard_query .= "$new_expr AS $alias";
										$coord_query .= "{$function}(distinct $alias) as $alias";
										#since we pushed a GB expression we need to update the ODKU clause
										$coord_odku[] = "$alias=VALUES($alias)";
								} else {
										$non_distrib_table = md5($alias.$this->uuid);

										$alias_cnt = trim($alias,'`');
										$alias_cnt = "`{$alias_cnt}_cnt`";
										$shard_query .= "({$base_expr}) AS {$alias}";		
										$coord_query .= "{$non_distrib_table}.`expr` AS {$alias}";

										#need to push a COUNT expression into the SELECT clause
										$push_select[] = "COUNT({$base_expr}) as {$alias_cnt}";
										$coord_odku[] = "{$alias_cnt}={$alias_cnt} +  VALUES({$alias_cnt})";

										#add this expression to the list of non-distributable agg functions
										$this->non_distrib[] = array($alias => $function);
								}
								$push_group[] = $pos + 1;
								$group_aliases[] = $alias;
								

							break;

							case 'COUNT':
								$used_agg_func=1;
								$base_expr=trim($base_expr,' ()');

								$expr_info=explode(" ", $base_expr);
								if(!empty($expr_info[0]) && strtolower($expr_info[0]) == 'distinct') {
									if($this->verbose) {
										echo "Detected a COUNT [DISTINCT] expression!\n";
									}
									unset($expr_info[0]);
									$new_expr = join(" ", $expr_info);
									$shard_query .= "$new_expr AS $alias";
									$coord_query .= "COUNT(distinct $alias) as $alias";
									$coord_odku[] = "$alias = VALUES($alias)";
									$group_aliases[] = $alias;
									$push_group[] = $pos + 1;
									
								} else {
									$shard_query .= "COUNT({$base_expr}) AS $alias";
									$coord_query .= "SUM($alias) AS $alias";
									$coord_odku[] = "$alias=$alias + VALUES($alias)";
								}
	
							break;
		
							default:
								$error[] = array('error_clause' => $clause['base_expr'],
										'error_reason' => 'Unsupported aggregate function');
										
				
							break;
						}
					} else {
						$group[] = $pos+1;
						$group_aliases[] = $alias;
						$shard_query .= $base_expr . ' AS ' . $alias;
						$coord_query .= $alias;
						$coord_odku[] = "$alias=VALUES($alias)";
					}
			
				break;
	
				case 'operator':
				case 'const':
				case 'colref':
				case 'reserved':
				case 'function':
					$this->orig_group[] = $alias;	
					$group[] = $pos+1;
					$group_aliases[] = $alias;
					$shard_query .= $base_expr . ' AS ' . $alias;
					$coord_query .= $alias;
					$coord_odku[] = "$alias=VALUES($alias)";
			
				break;
	
				default:
					$error[] = array('error_clause' => $clause['base_expr'],
							'error_reason' => 'Unsupported expression type (did you forget an alias on an aggregate expression?)');
				break;
			}	
			
		}

		$sql = "SELECT ";
		if($straight_join) $sql .= "STRAIGHT_JOIN ";

		$sql .= $shard_query;
		foreach($push_select as $clause) {
			$sql .= "," . $clause;
		}
	
		$shard_group=array();	
		#merge pushed and provided group-by
		if($used_agg_func) {
			$this->used_agg_func = true;
			$shard_group = $group;
			foreach($push_group as $push) {
				$shard_group[]=$push;
			}
			#they shouldn't conflict, but this ensures so
			$shard_group = array_unique($shard_group);
		} else {
			$group = array(); $shard_group = array();
		}

		#we can't send pushed group by to the coord shard, so send the expression based 
		return array('error'=>$error,'shard_sql'=>$sql, 'coord_odku' => $coord_odku, 'coord_sql'=> 'SELECT ' . $coord_query,'shard_group' => join(',',$shard_group), 'coord_group'=>join(',',$group), 'group_aliases' => join(',',$group_aliases));
	}
	
	protected function process_from($tables) {

		/* DEPENDENT-SUBQUERY handling
		*/

		$this->tables = array();

		foreach($tables as $key => $table) {
			if($table['table'] == 'DEPENDENT-SUBQUERY') {
			
			
			//	$this->process_sql($table['sub_tree']);
			//	$this->subqueries[$this->table_name] = $this->shard_sql;


				$sq = new OHIPShardQuery($this->shards, $this->params);
				
				$res = $sq->process_sql($table['sub_tree']);
		//		var_dump($res);
				//save
				$this->subqueries = $res['queries'];

				
				$sql = '(' . $this->coord_sql . ')';
				$tables[$key]['table'] = $sql;

			
		
				
				//call recusrive
			
				/*
				//build subqueries
				foreach($this->shard_sql as $q){
					$this->subqueries[$this->table_name] = array(array('sql' => $q , 'coord_odku' => $this->coord_odku, 'agg_key_cols' =>  $this->agg_key_cols));
				}*/
				
			//	$this->subqueries[$this->table_name] = array( array('sql' => $this->shard_sql , 'coord_odku' => $this->coord_odku, 'agg_key_cols' =>  $this->agg_key_cols) );
				//$this->subqueries[$this->table_name] = $this->shard_sql;

			} else {
				$this->tables[$table['table']] = 1;
			}
		}

		#escape the table name if it is unescaped
		if($tables[0]['alias'][0] != '`' && $tables[0]['alias'][0] != '(') $tables[0]['alias'] = '`' . $tables[0]['alias'] . '`';

		#the first table is always prefixed by FROM
		$sql = "FROM " . $tables[0]['table'] . ' AS ' . $tables[0]['alias'];
		$cnt = count($tables);

		#now create the rest of the FROM clause
		for($i=1;$i<$cnt;++$i) {

			if($tables[$i]['ref_type'] == 'USING') {
				$tables[$i]['ref_clause']="(" . trim($tables[$i]['ref_clause']) . ")";
			} elseif($tables[$i]['ref_type'] == 'ON') {
				$tables[$i]['ref_clause'] = ' (' . $tables[$i]['ref_clause'] . ") ";
			}

			if($sql) $sql .= " ";
			if($tables[$i]['alias'][0] != '`' && $tables[$i]['alias'][0] != '(') {
				$pos = strpos($tables[$i]['alias'], '.');
				if($pos !== false) {
					$info = explode('.', $tables[$i]['alias']);
					$table = $info[1];
					$tables[$i]['alias'] = '`' . $table . '`';	
				} else {
					$tables[$i]['alias'] = '`' . $tables[$i]['alias'] . '`';	

				}
			}
			$sql .= $tables[$i]['join_type'] . ' ' . $tables[$i]['table'] . ' AS ' . $tables[$i]['alias'] . ' ' . $tables[$i]['ref_type'] . $tables[$i]['ref_clause'];

		}
	
		return $sql;
	}

	function set_partition_info($column, $callback) {
		$this->partition_column = $column;
		if(!is_callable($callback)) {
			throw new Exception('Invalid callback (is_callable failed)');
		}

		$this->callback = $callback;
	}

	protected function get_partition_info($column, $key) {
		$result = call_user_func($this->callback, $column, $key);
	#	if($this->verbose) {
	#		echo "PARTITION LOOKUP: $column, $key => $result\n";
	#	}

		if(is_array($result)) {
			$keys=array_keys($result);
			$result=$keys[0];
		}
		return $result;
	}

	protected function append_all(&$queries, $append) {
		for($i=0;$i<count($queries);++$i) {
			$queries[$i] .= $append;
		}
	}

	protected function process_where($where) {
		$this->in_lists = array();
		$prev = "";
		$next_is_part_key = false;
		$this->force_shard = false;
		$shard_id = false;
		$this->force_broadcast = false;
		$total_count=0;

		$sql = "WHERE";
		$queries = array($sql);

		$start_count=count($where);
		foreach($where as $pos => $clause) {
			if(empty($where[$pos])) continue;
			$sql .= " ";
			$this->append_all($queries, " ");
			if($next_is_part_key) {
				if(!trim($clause['base_expr'])) continue;
				if($clause['expr_type'] == 'const' && $shard_id = $this->get_partition_info($prev, $clause['base_expr'])) {
					if($this->verbose) echo "PARTITION SELECTION SELECTED SHARD_ID: $shard_id\n";
					$this->force_shard = $shard_id;
				}
				$next_is_part_key = false;
			}

			if ($clause['expr_type'] == 'operator')  {
				if(strtolower($clause['base_expr']) == 'between' &&
				$this->between_opt && ($this->between_opt == '*' || $this->between_opt == $prev) ) {
					$offset=0;
					$operands=array();
					#find the operands to the between expression	
					$and_count=0;

					for($n=$pos+1;$n<$start_count;++$n) {
						if($where[$n]['expr_type'] == 'operator' && strtoupper($where[$n]['base_expr']) == 'AND') {
							if($and_count) {
								break;
							} else {
								$and_count+=1;
							}
						} 
						$operands[$offset]=array('pos'=>$n,'base_expr' => $where[$n]['base_expr']);
						++$offset;
					}

					#determine what kinds of operands are in use
					$matches = $vals = array();
					$is_date = false;

					//var_dump($operands);
					if(is_numeric(trim($operands[0]['base_expr'])) || preg_match("/('[0-9]+-[0-9]+-[0-9]+ [0-9]{2}:[0-9]{2}:[0-9]{2}')/", $operands[0]['base_expr'], $matches)) {
						if($matches) {
							$vals[0] = $matches[0];
							$matches = array();
					 		preg_match("/('[0-9]+-[0-9]+-[0-9]+ [0-9]{2}:[0-9]{2}:[0-9]{2}')/", $operands[2]['base_expr'], $matches);
							$vals[1] = $matches[0];
			
							$is_date =true;
						} else {
							$vals[0]  = $operands[0]['base_expr'];
							$vals[1]  = $operands[2]['base_expr'];

						}
						if(!$is_date) {
							$sub_tree = array();
							for($n=$vals[0];$n<=$vals[1];++$n) {
								$sub_tree[] = $n;
							}
						} else {

						
							//get table name
							$tables = array_keys($this->tables);
					
							//check date partitions
							$stmt = $this->mysqli->query("SHOW CREATE TABLE " . $tables[0]);
							if(!$stmt) {
								throw new Exception("While getting partition info ");
							}
							$result = $stmt->fetch_assoc();

							//get list of partitions for table
							preg_match_all('/PARTITION (?P<partition>.*) VALUES/', $result["Create Table"], $matches);
						
							//parse values to DateTime
							try{
								$minDateTime = new DateTime(str_replace("'", "", $vals[0]));
								$maxDateTime = new DateTime(str_replace("'", "", $vals[1]));
							} catch(Exception $e){
								//handle failure
							}
							
							//validate order
							if($minDateTime > $maxDateTime){
								//bad
								
							}
							
						//	echo 'Min';
						//	var_dump($minDateTime->format('Y-m-d H:i:s'));
						//	echo 'Max';
						//	var_dump($maxDateTime->format('Y-m-d H:i:s'));
					
							
							$partition_match_count = 0;
							$partition_matches = array();
							//loop existing partitions
							foreach($matches["partition"] as $p){
								//get the DateTime of the partition
								$partitionDateTime = DateTime::createFromFormat('pYmdH', $p);
								//check for number of partitions inside the range
								if(($minDateTime < $partitionDateTime) && ($partitionDateTime <= $maxDateTime)){
									//save	
									$partition_matches[] = $partitionDateTime;
									$partition_match_count++;
								}
							}
							
							//check for missing partitions
							if(count($matches["partition"]) == 0){
								$sub_tree[] = " '" . $minDateTime->format('Y-m-d H:i:s')  . "'  AND '" . $maxDateTime->format('Y-m-d H:i:s') . "'\n";
							}
							
							//no partitions mean keep original where
							if($partition_match_count == 0){
								$sub_tree[] = " '" . $minDateTime->format('Y-m-d H:i:s')  . "'  AND '" . $maxDateTime->format('Y-m-d H:i:s') . "'\n";
							}
							
							//build list to split on partitions
							$count = 0; 
							$prev = '';
							foreach($partition_matches as $p){
								
								//last
								if($count == $partition_match_count - 1){
									$sub_tree[] = " '" . $p->format('Y-m-d H:i:s') . "'  AND '" . $maxDateTime->format('Y-m-d H:i:s') . "'\n";	
								}
					
								//first
								if($count == 0){
									//subtract 1 sec
									$tmp = new DateTime($p->format('Y-m-d H:i:s'));
									date_sub($tmp, date_interval_create_from_date_string('1 second'));
									$sub_tree[] = " '" . $minDateTime->format('Y-m-d H:i:s')  . "'  AND '" . $tmp->format('Y-m-d H:i:s') . "'\n";
									
								}
							
								//middle 
								if(($count > 0) && ($count < $partition_match_count)){
									$tmp = new DateTime($p->format('Y-m-d H:i:s'));
									date_sub($tmp, date_interval_create_from_date_string('1 second'));
									$sub_tree[] = " '" . $prev->format('Y-m-d H:i:s') . "'  AND '" . $tmp->format('Y-m-d H:i:s') . "'\n";
								}
								
							
								$count++;
								$prev = $p;
							}
				

							#conversion of date between requires connecting
							#to the database to make sure that the date_diff calculation
							#is accurate for the timezone in which the database servers are
			/*	
							//$date_sql = "SELECT datediff(" . $vals[1] . ',' . $vals[0] . ") as `d`";
							$date_sql = "SELECT TIMESTAMPDIFF(HOUR," . $vals[0] . ',' . $vals[1] . ") as `d`";
							
							if($this->verbose) {
								echo "Sending SQL to do date calculation:\n$date_sql\n\n";
							}

							//query 
							$stmt = $this->mysqli->query($date_sql);
							if(!$stmt) {
								throw new Exception("While doing date diff: " . $this->my_error($this->conn));
							}
							$row = $stmt->fetch_assoc();
							
					
					
							$days = $row['d'];
							for($n = 0; $n < $days; ++$n) {
							
								$sub_tree[] = $vals[0] . " + interval $n hour AND ".$vals[0] . " + interval $n hour + interval 59 minute + interval 59 second";
							}
				*/		
						}
					//	var_dump($sub_tree);
				/*
					if(is_numeric(trim($operands[0]['base_expr'])) ||
					 preg_match("/('[0-9]+-[0-9]+-[0-9]+')/", $operands[0]['base_expr'], $matches)) {
						if($matches) {
							$vals[0] = $matches[0];
							$matches = array();
					 		preg_match("/('[0-9]+-[0-9]+-[0-9]+')/", $operands[2]['base_expr'], $matches);
							$vals[1] = $matches[0];
			
							$is_date =true;
						} else {
							$vals[0]  = $operands[0]['base_expr'];
							$vals[1]  = $operands[2]['base_expr'];

						}
						if(!$is_date) {
							$sub_tree = array();
							for($n=$vals[0];$n<=$vals[1];++$n) {
								$sub_tree[] = $n;
							}

						} else {

							#conversion of date between requires connecting
							#to the database to make sure that the date_diff calculation
							#is accurate for the timezone in which the database servers are
						
							$date_sql = "SELECT datediff(" . $vals[1] . ',' . $vals[0] . ") as `d`";
							if($this->verbose) {
								echo "Sending SQL to do date calculation:\n$date_sql\n\n";
							}

							$stmt = $this->my_query($date_sql);
							if(!$stmt) {
								throw new Exception("While doing date diff: " . $this->my_error($this->conn));
							}

							$row = mysql_fetch_assoc($stmt);
							$days = $row['d'];
							for($n=0;$n<=$days;++$n) {
								$sub_tree[] = $vals[0] . " + interval $n hour";
							}
						}
*/
						
						for($n = $pos+1; $n <= $operands[2]['pos']; ++$n) {
							unset($where[$n]);
						}

						if($this->verbose) {
							$this->messages[] = "A BETWEEN has been converted to an IN list with " . count($sub_tree) . " items\n";
						}
						$this->in_lists[] = $sub_tree;
						$old = $queries;

						$queries = array("");
						$ilist = "";
						$sub_tree=array_values($sub_tree);

						if(count($sub_tree) >= $this->inlist_merge_threshold ) {
							for($z=0;$z<count($sub_tree);++$z) {
								if($ilist) $ilist .= ",";
								$ilist .= $sub_tree[$z];
								if((($z+1) % $this->inlist_merge_size)  == 0 ) {
									foreach($old as $sql) {
										$queries[] = $sql . " IN (" . $ilist . ")";
									}
									$ilist= "";
								}
							}
							foreach($old as $sql) {
								if($ilist) $queries[] = $sql . " IN (" . $ilist . ")";
							}
							$ilist= "";
						} else {
					
							foreach($sub_tree as $val) {
								foreach($old as $sql) {
									$queries[] = $sql .= " BETWEEN $val";
								}
							}	
						}

						unset($sub_tree);

						continue;
					} else {
						if($this->verbose) {
							echo "BETWEEN could not be optimized - invalid operands\n";
						}
					}

				} elseif($clause['base_expr'] == '=' &&
					($this->partition_column && strtolower($this->partition_column) == strtolower($prev) && !$this->force_broadcast ))
			 	{
					if(!$this->force_shard) {
						$next_is_part_key=true;
					} else {
						if($this->verbose) {
							echo "More than one partition key found.  Query broadcast forced\n";
						}
						$this->force_shard = false;
						$this->force_broadcast = true;
					}
				}
				$this->append_all($queries, $clause['base_expr']);
			} elseif($clause['expr_type'] != 'in-list') {
				$this->append_all($queries, $clause['base_expr']);
				$prev = $clause['base_expr'];

			} elseif($this->inlist_opt && ($this->inlist_opt == '*' || $this->inlist_opt == $prev)) {
				$old = $queries;
				$queries=array();
				
				foreach($clause['sub_tree'] as $vals) {

					foreach($old as $sql) {
						$queries[] = "$sql ($vals) ";
					}
				}	

			} else {
				$prev = $clause['base_expr'];
				$this->append_all($queries, $prev);
			}
			
		}

		foreach($queries as $pos => $q) {
			if(!trim($q)) unset($queries[$pos]);
		}

		return array_values($queries);
	}

	/* if $sql is an Array(), then it is assumed it is already parsed */	
	public function process_sql($sql) {

		$this->sql = $sql;
		$parser = null;
		$straight_join = false;

		$this->shard_sql = ""; #identical SQL which will be broadcast to all shards
		$this->coord_sql = ""; #the summary table sql		
		$this->coord_odku = ""; #on duplicate key update sql
		
		$this->in_lists = array();
		$this->used_agg_func = false;
		$error = array();
		$select = null;
				
				
		//parse $sql using PHPSQLPaRSER
		if(!is_array($sql)) {
			#TODO: support parser re-use	
			#$this->parsed = $this->client->do('sql_parse',$sql);
			$parser = new PHPSQLParser($sql);	

			$this->parsed = $parser->parsed;
		} else {
			$this->parsed = $sql;
		}

		//var_dump($this->parsed);exit;
		//Handle UNION ALL
		if(!empty($this->parsed['UNION ALL'])) {
			$queries = array();
			
			//loop through each UNION ALL clause
			foreach($this->parsed['UNION ALL'] as $sub_tree) {
			
				//parsed hack
				if(isset($sub_tree[""]));
				unset($sub_tree[""]);

				$sq = new OHIPShardQuery($this->shards, $this->params);
				$res = $sq->process_sql($sub_tree);
	
				//save
				
				$coord_sql = $res['coord_sql'];
				$this->table_name = $res['table'];
				$this->non_distrib = $res['non_distrib'];
				$this->orig_group = $res['orig_group'];
				
				$queries = array_merge($queries, $res['queries']);
			}	
			
		//	$this->table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
		//	$coord_sql = "SELECT * from " . $this->table_name;
		
		//Handle UNION 
		} elseif(!empty($this->parsed['UNION'])) {
			$queries = array();
			foreach($this->parsed['UNION'] as $sub_tree) {
			
				$sq = new OHIPShardQuery($this->shards, $this->params);
				$res = $sq->process_sql($sub_tree);
	
				$queries = array_merge($queries, $res['queries']);
		
			//	$this->process_sql($sub_tree);
				//$queries = array_merge($queries,$this->shard_sql);
			}	
			$this->table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
			
			#UNION operation requires deduplication of the temporary table
			$coord_sql = "SELECT DISTINCT * from " . $this->table_name;
		} elseif(!empty($this->parsed['SELECT'])) {
		
			#reset the important variables	
			$select = $from = $where = $group = $order_by = "";
			$this->errors = array();	
			
			//Handle SELECT
			if(!empty($this->parsed['OPTIONS']) && in_array('STRAIGHT_JOIN', $this->parsed['OPTIONS'])) { 
				$straight_join=true;
				unset($this->parsed['OPTIONS']);
			}

			$select = $this->process_select($this->parsed['SELECT'], $straight_join);

			if(!empty($select['error'])) {
				$this->errors = $select['error'];
				return false;
			}
			unset($this->parsed['SELECT']);

			//Handle FROM
			if(empty($this->parsed['FROM'])) {
				$this->errors = array('Unsupported query', 'Missing FROM clause');
				return false;
			} else {
				$select['shard_sql'] .= "\n" . $this->process_from($this->parsed['FROM']);
			
			
				//check if table(s) exist
				$in_tables = "";
				foreach(array_keys($this->tables) as $val) {
					if($in_tables !== "") $in_tables .= ",";
					$in_tables .= "'{$val}'";
				}

				$sql = "select count(*) cnt from information_schema.columns where table_schema='" . 
					$this->tmp_shard['db'] . "' and column_name='" . $this->partition_column .
					"' and table_name IN({$in_tables});";
				
				//var_dump($sql); exit;
			/*	
				//query
				$stmt = $this->mysqli->query($sql);
				$row = $stmt->fetch_assoc();				
				if($row['cnt'] === '0') {
					$this->force_shard = array_rand($this->shards);
				}			
*/
				$this->table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
				
		/*		if(empty($this->parsed['GROUP'])) {
					if( !empty($this->non_distrib) ) {
						$select['coord_sql'] .= "\nFROM dual2";
					} else {
						$select['coord_sql'] .= "\nFROM `$this->table_name`";
					}
				} else {
					$select['coord_sql'] .= "\nFROM `$this->table_name`";
				}
				*/
				
				$select['coord_sql'] .= "\nFROM `$this->table_name`";
				
	
				#we only select from a single table here	
				if(!empty($this->non_distrib)) {
					$select_group = "";
					foreach($this->orig_group as $gb) {
						if(is_array($gb)) continue;
						if(trim($gb) === "") continue;
						if($select_group) $select_group .= ",";
						$select_group .= $gb;
						
					}
					if($select_group) $select_group .= ",";
					foreach($this->non_distrib as $alias => $function) {
					
						
						$alias = array_pop(array_keys($function));	
						$function = array_pop($function);
						
						
							

						#$select['coord_sql'] .= "\nNATURAL JOIN (select {$select_group}{$function}(`val`) AS `expr` FROM " . md5($alias);
						
						if(preg_match('/90TILE/', $function)){
							$select['coord_sql'] .= "\nJOIN (select (`val`) AS `expr` FROM " . md5($alias.$this->uuid);
						}else{
							$select['coord_sql'] .= "\nJOIN (select {$select_group}{$function}(`val`) AS `expr` FROM " . md5($alias.$this->uuid);
						}
						
						
						if($select_group) $select['coord_sql'] .= " GROUP BY " . trim($select_group,',');
						
							//percentile hack
						if(preg_match('/90TILE/', $function)){
					
							$select['coord_sql'] .= ' ORDER BY `val` desc LIMIT $count , 1';
						}
						
						$select['coord_sql'] .= ") as " .  md5($alias.$this->uuid);
						if($select_group) $select['coord_sql'] .= ' USING(' . trim($select_group,',') . ')';
						
						
					}
				}	
				unset($this->parsed['FROM']);
			}

			//Add additional WHERE conditions
			if($this->push_where !== false && $this->push_where) {
				if(!empty($this->parsed['WHERE'])) {
					$this->parsed['WHERE'][] = array('expr_type' => 'operator', 'base_expr' => 'and','sub_tree' => '');
				}
				if(!$parser) $parser = new PHPSQLParser();
				$this->messages[] = "Where clause push detected.  Pushing additional WHERE condition:'" . $this->push_where . "' to each storage node.\n";
				if($this->push_where) foreach($parser->process_expr_list($parser->split_sql($this->push_where)) as $item) $this->parsed['WHERE'][]=$item;

			} 

			//Handle WHERE
			#note that this will extract inlists and store them in $this->in_lists (if inlist optimization is on) 	
			if(!empty($this->parsed['WHERE'])) {
				$where_clauses = $this->process_where($this->parsed['WHERE']);
				unset($this->parsed['WHERE']);
			}
		
			//Handle GROUP BY 
			if(!empty($this->parsed['GROUP'])) {
				foreach($this->parsed['GROUP'] as $gb) {
					if($select['shard_group'] !== "") $select['shard_group'] .= ',';
					$select['shard_group'] .= $gb['base_expr'];
				}
			}
			unset($this->parsed['GROUP']);

			//Handle ORDER BY
			if(!empty($this->parsed['ORDER'])) {
				$order_by = "";
				foreach($this->parsed['ORDER'] as $o) {
					if($order_by) $order_by .= ",";
					$order_by .= $o['base_expr'] . ' ' . $o['direction'];	
				}	
				$order_by = "ORDER BY {$order_by}";
				unset($this->parsed['ORDER']);
			}

			//Handle LIMIT
			if(!empty($this->parsed['LIMIT']) ) {
				$order_by .= " LIMIT {$this->parsed['LIMIT']['start']},{$this->parsed['LIMIT']['end']}";
				unset($this->parsed['LIMIT']);
			}

			//check for left overs
			foreach($this->parsed as $key => $clauses) {
				$this->errors[] = array('Unsupported query',$key . ' clause is not supported');
			}

			//exit on errors
			if($this->errors) {
				return false;
			}

			#process_select only provides a list of positions
			if($select['coord_group']) $select['coord_group'] = ' GROUP BY ' . $select['coord_group'];
			if($select['shard_group']) $select['shard_group'] = ' GROUP BY ' . $select['shard_group'];
			
			//build queries to be sent to shards
			$queries = array();
			if(!empty($where_clauses)) {
				foreach($where_clauses as $where) {
					$queries[] = $select['shard_sql'] . ' ' . $where . ' ' . $select['shard_group'] . ' ORDER BY NULL';
				}
			} else {
				$queries[] = $select['shard_sql'] . $select['shard_group'] . ' ORDER BY NULL';
			}

		} elseif(!empty($this->parsed['INSERT'])) { 

			if($this->parsed['INSERT']['cols'] == 'ALL') {
				$this->errors = array('Unsupported query', 'INSERT statements must include a column list');
				return false;
			}

			$shard_key_pos = false;	
			$col_list="";
			foreach($this->parsed['INSERT']['cols'] as $pos => $col) {

				if($col_list) $col_list .= ", ";
				$col_list .= $col;
				if(trim($col) == $this->partition_column) {
					$shard_key_pos = $pos;
				}	
			}

			$sql = "INSERT INTO `" . $this->parsed['INSERT']['table'] . "` ({$col_list}) VALUES ";
			$values = array();
			$val_count = 0;
			
			foreach($this->parsed['VALUES'] as $_vals) {

				$vals = trim($_vals,'()');
				$vals = explode(',',$vals);

				#this row is going to go to a single shard
				if($shard_key_pos !== false) {

					$shard_id=$this->get_partition_info($this->partition_column, $vals[$shard_key_pos]);

					if(empty($values[$shard_id])) {
						$values[$shard_id] = $_vals;	
					} else {
						$values[$shard_id] .= ",{$_vals}";
					}

					$val_count++;

				#the row is going to all shards
				} else {
					foreach($this->shards as $shard_id => $shard) {
						if(empty($values[$shard_id])) {
							$values[$shard_id] = $_vals;	
						} else {
							$values[$shard_id] .= ",{$_vals}";
						}

						$val_count++;
					}

				}
			}

			$this->insert_sql = $sql;
			$this->insert_values = $values;

			return true;


		} elseif(!empty($this->parsed['UPDATE'])) { 
			$this->errors = array('Unsupported query', 'Missing expected clause:SELECT');
			return false;

		} elseif(!empty($this->parsed['DELETE'])) { 
			$this->errors = array('Unsupported query', 'Missing expected clause:SELECT');
			return false;

		} elseif(!empty($this->parsed['REPLACE'])) { 
			$this->errors = array('Unsupported query', 'Missing expected clause:SELECT');
			return false;

		} else {
			$this->errors = array('Unsupported query', 'Missing expected clause:SELECT');
			return false;
		}

		#if($order_by == "") $order_by = "ORDER BY NULL";
		if(isset($coord_sql))
			$this->coord_sql = $coord_sql;
		else
			$this->coord_sql = $select['coord_sql'] . ' ' . $select['coord_group'] . ' ' . $order_by;
		

		if($this->coord_odku === "") 
			$this->coord_odku = $select['coord_odku'];
		
		
		if(!isset($this->agg_key_cols))
		$this->agg_key_cols = $select['group_aliases'];
		
		
		//rebuild shard arry to include odku/keys
		$shard_sql = array();
		foreach($queries as $q){
			if(!is_array($q))
				$shard_sql[] = array('sql' => $q , 'coord_odku' => $this->coord_odku, 'agg_key_cols' =>  $this->agg_key_cols);
			else
				$shard_sql[] = $q;
		}
		$this->shard_sql = $shard_sql;
			
		
		
		if ($this->verbose) {
			echo "-- INPUT SQL:\n$sql\n";
			echo "\n--PARALLEL OPTIMIZATIONS:\n";


			if($this->agg_key_cols) { 
				echo "\n* The following projections were selected for a UNIQUE CHECK on the storage node operation:\n{$this->agg_key_cols}\n";
			        if($this->coord_odku) echo "\n* storage node result set merge optimization enabled:\nON DUPLICATE KEY UPDATE\n" . join(",\n",$this->coord_odku) . "\n";
			}
			echo "\n";

			foreach($this->messages as $msg) {
				echo "-- $msg\n";
			}

			echo "\n-- SQL TO SEND TO SHARDS:\n";
			print_r($this->shard_sql);
			echo "\n-- AGGREGATION SQL:\n{$this->coord_sql}" . ( $this->agg_key_cols && $this->coord_odku ? "\nON DUPLICATE KEY UPDATE\n" . join(",\n",$this->coord_odku) . "\n" : "\n");
		}

		return array('queries' => $this->shard_sql, 'sub_queries' => $this->subqueries, 'table' => $this->table_name, 'coord_sql' => $this->coord_sql, 'non_distrib' => $this->non_distrib, 'orig_group' => $this->orig_group);
	
	}

	protected function create_gearman_set($queries, $table, $complete_cb=null,$failure_cb=null, &$set) {
		$sets = array();

		if(!$set) 
			$set = new Net_Gearman_Set();
		

	//		var_dump($queries);
		//loop queries
	//	for($i = 0; $i < count($queries); $i++) {
		
		foreach($queries as $q){
	///	var_dump($queries[$i]); exit;
			foreach(($this->force_shard != false) ? array($this->shards[$this->force_shard]) : $this->shards as $shard) {
			

/*
				$task = new Net_Gearman_Task($this->fetch_method, array(
								'engine'=>$this->engine,
								'table_name' => $table, 
								'sql' => $queries[$i],
								'coord_odku'=> $this->coord_odku, 
								'shard'=>$shard,
								'tmp_shard'=>$this->tmp_shard,
								'agg_key_cols'=> $this->agg_key_cols,
								'when' => microtime(true))
							);
	*/

				$task = new Net_Gearman_Task($this->fetch_method, array(
							'engine'=> $this->engine,
							'table_name' => $table, 
							'sql' => $q['sql'],
							'coord_odku'=> $q['coord_odku'], 
							'shard'=> $shard,
							'tmp_shard'=> $this->tmp_shard,
							'agg_key_cols'=> $q['agg_key_cols'],
							'when' => microtime(true))
						);
				$task->attachCallback($complete_cb,Net_Gearman_Task::TASK_COMPLETE);	
				$task->attachCallback($failure_cb,Net_Gearman_Task::TASK_FAIL);
				$set->addTask($task);
			}
			//add
			$sets[] = $set;
		}	
		return $sets;
	}

	#runs a set of Gearman workers and waits for them to complete
	protected function run_set($set) { 

		if(!is_array($set)) $set = array($set);

		#start the workers about their work
		$spins = 0;
		$enters = 0;
		$total_sleep_time=0;
		$sleeps=0;
		if($this->verbose) echo "run_set: received another set of jobs\n";
		foreach($set as $the_set) {	
		
		//	var_dump($the_set);
		
			/* PHP isn't threaded.  When the asynch gearman callback happens, the change in workers is atomic*/
			$this->workers++;
			if($this->verbose) echo "run_set: Starting another gearman job\n";

			$this->client->runSet($the_set);

			/* shouldnt need to do this spinns
			
			/* we have to wait if there are currently $this->max_degree workers running 
			$entered=0;
			while($this->workers >= $this->max_degree) {
				if(!$entered) {
					$entered_at = microtime(true);
					$entered = 1;
					++$enters;
				}

				#we waste up to 2 seconds of wall time here, before we start sleeping for .01 seconds each spin
				if((($start_sleep = microtime(true)) - $entered_at) >= 2.0) {
					++$sleeps;
					sleep(.01);
					$total_sleep_time += microtime(true) - $start_sleep;
				} else {
					++$spins;
				}
				
			} 
			*/
			
		}

		if($this->verbose) echo "run_set: finished set execution.  enters: $enters, spins: $spins, sleeps: $sleeps, total_sleep_time: $total_sleep_time\n";

	}

	protected function aggregate_result() {
		
		//$this->my_select_db($this->tmp_shard['db'], $this->conn) or die($this->my_error($this->conn));
/*
		if($this->debug) {
			$stmt = $this->my_query( "select * from `{$this->table_name}`", $this->conn);
			if(!$stmt) throw new Exception($this->my_error() . "\n");
			while($row = $this->my_fetch_assoc($stmt)) {
				print_r($row);
			}
		}
*/
	
		#Now that the workers have completed, we have to send the final query
		#that returns the accumulated result, but first we need to decode the runlength encoded
		#values for the non-distributable aggregate functions

		$push_from = array();

		if(!empty($this->non_distrib)) {
		/*
			$sql = 'CREATE TEMPORARY TABLE dual2(x varchar(32)) engine=MYISAM;';

			
			
			$sql = 'CREATE TABLE dual2(x varchar(32)) engine=MYISAM;';
			
			
			mysql_query($sql, $this->conn) or die('could not create dual2: ' . mysql_error() . "\n");
			$sql = 'INSERT INTO dual2 values("X");';
			mysql_query($sql, $this->conn) or die('could not fill dual2: ' . mysql_error() . "\n");
	*/
		}
		foreach($this->non_distrib as $alias => $function) {   
			$alias = array_pop(array_keys($function));
			$function = array_pop($function);
			$table =  md5($alias.$this->uuid);
			$alias_cnt = "`" . trim($alias,'`') . '_cnt' . "`";   

			$select_group = "";
			foreach($this->orig_group as $gb) {
				if(is_array($gb)) continue;
				if(trim($gb)==="") continue;
				if($select_group !== "") $select_group .= ",";
				$select_group .= $gb;
			} 
			if($select_group) $select_group .= ",";
		//	mysql_query('set storage_engine=MYISAM', $this->conn);
			$sql = "CREATE TEMPORARY TABLE {$table} as select {$select_group}{$alias} as `val` from {$this->table_name} limit 0";
			$sql = "CREATE TABLE {$table} as select {$select_group}{$alias} as `val` from {$this->table_name} limit 0";
		
			//create temp non distrib table
			//mysql_query($sql, $this->conn) or die('Could not create temp table ' . mysql_error() . "\n");
			$this->mysqli->query($sql);

			//get values to fold out
			$sql = "select {$select_group}{$alias} as `val`, {$alias_cnt} as `cnt` from {$this->table_name}";
			#if($select_group !== "") {
			#	$sql .= " GROUP BY " . trim($select_group, ',');
			#}
	
			 
			//query 
			$stmt = $this->mysqli->query($sql);

			//check error
			if(!$stmt) {
				echo 'non distrib select error';
			}
			
	//		$sql = "SET @@global.max_allowed_packet = 104857600";
	//		if(!$this->mysqli->query($sql)){
	//				echo 'failed setting max_allowed_packet';
	//		}
			
			//loop results
			$count = 0;
			while($row = $stmt->fetch_assoc()) {
			
				$sql = "INSERT INTO {$table} VALUES ";
				$values = "";
				for($i=0;$i< $row['cnt'];++$i) {
					if($values !== "") $values .= ',';
					$values .= "(";
					$rowvals = "";	
					foreach($row as $col => $val) {
						if($col === 'cnt') continue;
						if($rowvals !== "") $rowvals .= ",";
						$rowvals .= "'" . $val . "'";
						$count++;
					}
					$values .= $rowvals . ')';
				}
				$sql .= $values;
				
				
				//execute and check for failure
				if(!$this->mysqli->query($sql)){
					echo 'failed inserting into temporary table';
				}
			}
			//echo 'COUNT: ' . $count ;
			
			$this->coord_sql = str_replace('$count', floor($count * .10), $this->coord_sql);
			
	//		if(count($this->non_distrib) == 1)
	//			$this->coord_sql .= " LIMIT 1";
		}
		
		//final aggreate query
		if($this->verbose){
			echo 'COORD SQL:\n';
			var_dump($this->coord_sql);
		}
		
		$stmt = $this->mysqli->query($this->coord_sql);

		//clean up temp tables
		$sql = "DROP TABLE IF EXISTS {$this->table_name};";
		if(!$this->mysqli->query($sql)) {
			throw new Exception(mysql_error($this->conn));
		}

		$tables = array_keys($this->subqueries);
		foreach($tables as $table) {
			$sql = "DROP TABLE IF EXISTS {$table};";
			if(!$this->mysqli->query($sql)) {
				throw new Exception(mysql_error($this->conn));
			}
		}
		
		foreach($this->non_distrib as $alias => $function) {   
			$table = md5($alias.$this->uuid);
			
			$sql = "DROP TABLE IF EXISTS {$table};";
			if(!$this->mysqli->query($sql)) {
				throw new Exception(mysql_error($this->conn));
			}
		}

		return $stmt;	
	}

	#callback that records that one of the workers failed	
	function handle_failure($func) {
		//decrement
		--$this->workers;
		if($this->verbose) {
			echo "WORKER_FAIL: handle_failure registered unsuccessful worker completion.\n";
		}
		if($this->verbose) {
			print_r($func);
			exit;
		}
		if(!$this->errors) 
			$this->errors = array();
		$this->errors[] = $func;
		
	}

	#this only works with fetch_resultset
	function generic_rowset($func, $handle, $result) {
		--$this->workers;
		if($result['err'] != NULL) {
			throw new Exception('ERROR: ' . $result['err'] . "\n");
		}
		foreach($result->rows as $row) {
			print_r($result);
		}
	}
	
	function collect_rowset($func, $handle, $result) {
		//decrement
		--$this->workers;

		if(is_array($result) && !empty($result['error'])) {
			$this->errors[] = $result['error'];
		}

	}
}
